package net.bwie.realtime.dwd.douyin.logs.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class IsNotFanFunction extends KeyedProcessFunction<String,String,String> {
    private transient ValueState<Integer> isNotFanTeam;

    @Override
    public void open(Configuration parameters) throws Exception {
        isNotFanTeam=
                getRuntimeContext().getState(new ValueStateDescriptor<Integer>("is_not_fan",Integer.class));
    }

    @Override
    public void processElement(String s,Context ctx, Collector<String> out) throws Exception {
        //解析json数据
        JSONObject jsonObject = JSON.parseObject(s);

        //获取事件类型
        String event_type = jsonObject.getString("event_type");
        if (event_type!=null && event_type.equals("加入粉丝团")){
            if (isNotFanTeam.value()==null){
                jsonObject.put("is_not_fan",1);
                isNotFanTeam.update(1);
            }else {
                jsonObject.put("is_not_fan",0);
            }
        }
        out.collect(jsonObject.toJSONString());
    }
}
